package core

import (
	"encoding/json"
	"errors"
	"fmt"
	"gark-elastic-job/engine"

	logger "github.com/sirupsen/logrus"
)

type JobManager struct {
	node *JobNode
}

func NewJobManager(node *JobNode) *JobManager {
	logger.Debug("start Job Manager")
	jobManager := &JobManager{node: node}
	go jobManager.watchJobConf()
	go jobManager.loopLoadJob()
	return jobManager
}

func (jobManager JobManager) AddJob(jobConf engine.GarkJobConf) error {
	// 判断集群是否存在
	resp, err := jobManager.node.etcd.Get(GroupConfPath + jobConf.Group)
	if err != nil {
		return err
	}
	if len(resp) == 0 {
		return errors.New("job group does't exist")
	}
	// 添加任务至etcd指定目录下面
	// 注意这里只能新增,不能修改；如果之前存在值的话,就会提示已经存在
	contents, err := json.Marshal(jobConf)
	if err != nil {
		return err
	}
	key := fmt.Sprintf("%s%s", JobConfPath, jobConf.Id)
	success, old, err := jobManager.node.etcd.PutNotExist(key, string(contents))
	if err != nil {
		return err
	}
	if !success {
		logger.Infof("key %s, previous value: %s", key, old)
		return errors.New("fail to create job")
	}
	return nil
}

func (jobManager JobManager) ListJob() ([]engine.GarkJobConf, error) {
	keyResp, valueResp, err := jobManager.node.etcd.GetWithPrefixKey(JobConfPath)
	if err != nil {
		return nil, err
	}

	if len(keyResp) == 0 {
		return nil, nil
	}

	jobConfList := make([]engine.GarkJobConf, 0)

	for _, value := range valueResp {
		jobConf := &engine.GarkJobConf{}
		err := json.Unmarshal(value, jobConf)
		if err != nil {
			logger.Warnf("failed to decode job conf, err: %v", err)
			continue
		}
		jobConfList = append(jobConfList, *jobConf)
	}

	return jobConfList, nil
}

// watchJobConf 监控作业配置节点的路径地址
func (jobManager JobManager) watchJobConf() {
	resp := jobManager.node.etcd.WatchKeyWithPrefix(JobConfPath)

	// 监听处理每个key变化的事件
	for keyChangeEvent := range resp.Event {
		jobManager.handleJobConfChangeEvent(keyChangeEvent)
	}
}

// 注意: 应该先loadGroup才能loadJob; 从逻辑上讲，应该是先有group才有job
// 设置一个重试的机制，可以避免此类问题
func (jobManager JobManager) loopLoadJob() {
	keys, valueResp, err := jobManager.node.etcd.GetWithPrefixKey(JobConfPath)
	if err != nil {
		logger.Errorf("failed to get job conf, err: %v", err)
		return
	}
	if len(keys) == 0 {
		logger.Infof("no job conf to be load")
		return
	}
	for _, value := range valueResp {
		jobConf := &engine.GarkJobConf{}
		err := json.Unmarshal(value, jobConf)
		if err != nil {
			logger.Infof("failed to decode job conf, err: %v", err)
			continue
		}
		// 将作业创建事件推送至scheduler进行调度
		jobManager.node.Scheduler.pushJobChangeCreate(engine.JobChangeEvent{
			Type: engine.JobCreateChangeEvent,
			Conf: *jobConf,
		})
	}

}

// handleJobConfChangeEvent 处理每个jobConfChangeEvent
func (jobManager JobManager) handleJobConfChangeEvent(event *engine.KeyChangeEvent) {
	switch event.Type {
	case engine.KeyCreateChangeEvent:
		jobManager.handleJobConfCreateEvent(event.Value)
	case engine.KeyUpdateChangeEvent:
		jobManager.handleJobConfUpdateEvent(event.Value)
	case engine.KeyDeleteChangeEvent:
		jobManager.handleJobConfDeleteEvent(event.Key)
	}
}

func (jobManager JobManager) handleJobConfCreateEvent(value []byte) {
	jobConf := engine.GarkJobConf{}

	err := json.Unmarshal(value, &jobConf)
	if err != nil {
		logger.Infof("failed to decode job conf, err: %v", err)
		return
	}

	// 将作业创建事件推送至scheduler进行调度
	jobManager.node.Scheduler.pushJobChangeCreate(engine.JobChangeEvent{
		Type: engine.JobCreateChangeEvent,
		Conf: jobConf,
	})
}

func (jobManager JobManager) handleJobConfUpdateEvent(value []byte) {
	// todo
}

func (jobManager JobManager) handleJobConfDeleteEvent(key string) {
	// todo
}

func (jobManager JobManager) AddGroup(groupConf engine.GroupConf) error {
	bytes, err := json.Marshal(groupConf)
	if err != nil {
		return err
	}

	success, _, err := jobManager.node.etcd.PutNotExist(GroupConfPath+groupConf.Name, string(bytes))
	if err != nil {
		return err
	}

	if !success {
		return errors.New("group already exist")
	}
	return nil

}

func (jobManager JobManager) ListGroup() ([]engine.GroupConf, error) {
	keyResp, valueResp, err := jobManager.node.etcd.GetWithPrefixKey(GroupConfPath)
	if err != nil {
		return nil, err
	}

	if len(keyResp) == 0 {
		return nil, nil
	}

	groupConfList := make([]engine.GroupConf, 0)
	for _, value := range valueResp {
		groupConf := &engine.GroupConf{}
		err := json.Unmarshal(value, groupConf)
		if err != nil {
			logger.Warnf("decode job conf err: %v", err)
			continue
		}
		groupConfList = append(groupConfList, *groupConf)
	}

	return groupConfList, nil
}
